package com.gxy.learn.rmq.consumer.oms;

import io.openmessaging.Message;
import io.openmessaging.MessagingAccessPoint;
import io.openmessaging.OMS;
import io.openmessaging.OMSBuiltinKeys;
import io.openmessaging.consumer.PullConsumer;
import io.openmessaging.producer.Producer;
import io.openmessaging.producer.SendResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 用OMS PullConsumer 来从指定的队列中拉取消息
 */
public class OmsPullConsumer {
    private final static Logger log = LoggerFactory.getLogger(OmsPullConsumer.class);

    public static void main(String[] args) {
        final MessagingAccessPoint messagingAccessPoint = OMS.getMessagingAccessPoint("oms:rocketmq://localhost:9876/topic-oms");
        messagingAccessPoint.startup();
        final Producer producer = messagingAccessPoint.createProducer();
        final PullConsumer consumer = messagingAccessPoint.createPullConsumer(OMS.newKeyValue().put(OMSBuiltinKeys.CONSUMER_ID, "OMS_CONSUMER"));
        messagingAccessPoint.startup();
        log.info("MessagingAccessPoint startup OK%n");
        final String queueName = "TopicTest";
        producer.startup();
        Message msg = producer.createBytesMessage(queueName, "Hello Open Messaging".getBytes());
        SendResult sendResult = producer.send(msg);
        log.info("Send Message OK. MsgId: {}}", sendResult.messageId());
        producer.shutdown();
        consumer.attachQueue(queueName);
        consumer.startup();
        log.info("Consumer startup OK");
        // 运行直到发现一个消息被发送了
        boolean stop = false;
        while (!stop) {
            Message message = consumer.receive();
            if (message != null) {
                String msgId = message.sysHeaders().getString(Message.BuiltinKeys.MESSAGE_ID);
                log.info("Received one message: {}", msgId);
                consumer.ack(msgId);
                if (!stop) {
                    stop = msgId.equalsIgnoreCase(sendResult.messageId());
                }
            } else {
                log.info("Return without any message");
            }
        }
        consumer.shutdown();
        messagingAccessPoint.shutdown();
    }
}